package com.teamsun.mqtt.service;

import io.netty.buffer.ByteBuf;
import io.netty.util.ReferenceCountUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Stream;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.entity.vo.DelaySnedMsg;
import com.teamsun.mqtt.entity.vo.DelayWillRemoveMsg;
import com.teamsun.mqtt.entity.vo.SendableMsg;

/**
 * 重发机制服务
 * 
 * @author tzj
 * 
 */
public class ReSendService {

	Logger logger = Logger.getLogger(getClass());
	/**
	 * 延迟队列
	 */
	DelayQueue<DelaySnedMsg> delayQueue = new DelayQueue<DelaySnedMsg>();

	/**
	 * 用于映射客户端标识和延迟对象
	 */
	ConcurrentMap<String, DelaySnedMsg> concurrentMap = new ConcurrentHashMap<String, DelaySnedMsg>();

	/**
	 * 客户端与 “客户端 消息”这样的组的映射
	 */
	ConcurrentHashMap<String, BlockingQueue<String>> clentid2clientAndMsgs = new ConcurrentHashMap<String, BlockingQueue<String>>();

	
	
	@Autowired
	MessagePushService messagePushService;
	
	/**
	 * 数据库操作线程组
	 */
	ExecutorService dboptgroup;

	 int  resendtimes=3;

	public ReSendService() {
		start();
	}

	
	/**
	 * 提交一个 延迟发送对象到 延迟队列
	 * 
	 * 至于延迟对象的 buff 首先在map里面找，如果能找到就标记保留+1 标记的时候有可能异常 在异常里面新建buff给延迟对象
	 * 
	 * @param delaySnedMsg
	 */

	public void submitDelayMsg(DelaySnedMsg delaySnedMsg) {

		SendableMsg sendableMsg = delaySnedMsg.getSendableMsg();
		String clientid = sendableMsg.getRecvclientid();
		String  cAndmid=clientid+":"+sendableMsg.getMessageid();
		ByteBuf willSendBytebuff = sendableMsg.getMsgContent().retainedDuplicate();
		sendableMsg.setDupTimes(sendableMsg.getDupTimes()+1);
		
		if (willSendBytebuff.refCnt() <= 0) {
			logger.error("消息已经被释放,添加延迟队列失败");
			return;
		}

		delayQueue.offer(delaySnedMsg);
		DelaySnedMsg delaySnedMsglod = concurrentMap.put(cAndmid, delaySnedMsg);
		BlockingQueue<String> blockingQueue = null;
		if (!clentid2clientAndMsgs.containsKey(clientid)) {
			blockingQueue = new LinkedBlockingQueue<String>();
			clentid2clientAndMsgs.put(clientid, blockingQueue);
		} else
			blockingQueue = clentid2clientAndMsgs.get(clientid);

		if (blockingQueue != null && delaySnedMsglod == null)
			blockingQueue.offer(cAndmid);

	}

	/**
	 * 启动一个线程重新发送消息给客户端
	 */
	public void start() {

		new Thread(()->{
			
			while (true) {
				DelaySnedMsg delaySnedMsg = null;
				SendableMsg sendableMsg = null;
				try {
					delaySnedMsg = delayQueue.take();
					if(delaySnedMsg.isHasSend())
						continue;
					sendableMsg = delaySnedMsg.getSendableMsg();
				} catch (InterruptedException e) {
					logger.debug(e);
				}

				if (delaySnedMsg != null && sendableMsg != null) {

					if (sendableMsg.getDupTimes() >= resendtimes) {// 3
						// 应该可配置
						SendMsg unsceesssMsg = new SendMsg(sendableMsg
								.getRecvclientid(), sendableMsg
								.getMessageid(),
								messagePushService.MSG_SENDUNSUCCESS,
								sendableMsg.getSendclientid());
						messagePushService.delayOCalQueue.offer(new DelayWillRemoveMsg(unsceesssMsg));
						ReferenceCountUtil.release(sendableMsg.getMsgContent());
						logger.debug(sendableMsg.getSendclientid()+"重发给"+sendableMsg.getRecvclientid()+"三次不成功，加入成功队列");
						sendableMsg=null;
					}

					else {
					   if(sendableMsg.getMsgContent().refCnt()<=0){
						   logger.warn("消息已经被释放,延迟发送失败");
					   }else{
						   logger.debug( "第"+sendableMsg.getDupTimes()+"次给"+sendableMsg.getSendclientid()+"重新发送给"+sendableMsg.getRecvclientid()+
								   "消息主题:"+sendableMsg.getTopname());
						
							messagePushService.sendMsg2Client(sendableMsg);
					   }
					}

				}

			}
			
		}).start();;
		
	}

	/**
	 * 返回演出消息对象
	 * 
	 * @param clientid
	 * @param msgid
	 * @return
	 */
	public DelaySnedMsg getDelaySnedMsg(String clientid, Integer msgid) {
		String cAndmid = clientid + ":" + msgid;
		if (concurrentMap.containsKey(cAndmid)) {
			return concurrentMap.get(cAndmid);
		}
		return null;
	}

	/**
	 * 根据客户端标识移除 延迟队列消息 首先根据客户端标识找到 客户端+消息 队列 遍历 客户端+消息 队列 得到每一个延迟消息 从 延迟队列移除它
	 * 客户端+消息 队列 根据客户端标识移除键 clentid2clientAndMsgs
	 * 
	 * @param clentid
	 * @param recvsuccess
	 *            是否是成功接收了 客户端标识
	 * @return 成功与否
	 */
	public boolean remmoveFromClientId(final String clentid, boolean recvsuccess) {

		
		logger.debug("删除延迟队列客户端" + clentid + "的消息");
		
	
		if (clentid2clientAndMsgs.containsKey(clentid)) {
			BlockingQueue<String> blockingQueue = clentid2clientAndMsgs
					.get(clentid);
			List<String> strings = new ArrayList<String>();
			blockingQueue.drainTo(strings);
			
			
			Stream<String> stream=strings.parallelStream();
			
			stream.forEach((String string)->{
				
				DelaySnedMsg delaySnedMsg = concurrentMap.get(string);

				if (delaySnedMsg != null) {

					if (!recvsuccess) {
						SendableMsg sendableMsg = delaySnedMsg.getSendableMsg();
						SendMsg msg = new SendMsg(
								sendableMsg.getRecvclientid(),
								sendableMsg.getMessageid(),
								messagePushService.MSG_SENDUNSUCCESS,
								sendableMsg.getSendclientid());
						messagePushService.delayOCalQueue.offer(new DelayWillRemoveMsg(msg));
					}

					delayQueue.remove(delaySnedMsg);
				}
				concurrentMap.remove(string);
			});
			
			stream.close();
			blockingQueue = null;
			clentid2clientAndMsgs.remove(clentid);

		}

		return false;
	}

	public void setResendtimes(int resendtimes) {
		this.resendtimes = resendtimes;
	}



	public ExecutorService getDboptgroup() {
		return dboptgroup;
	}

	public void setDboptgroup(ExecutorService dboptgroup) {
		this.dboptgroup = dboptgroup;
	}
	
}
